package com.shujia.flink.source

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala._

object Demo3MyqlSource {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    //使用自定义source
    val mysqlDS: DataStream[AdminCode] = env.addSource(new MysqlSource)

    val countDS: DataStream[(String, Int)] = mysqlDS
      .map(ac => (ac.cituName, 1))
      .keyBy(_._1)
      .sum(1)

    countDS.print()

    env.execute()
  }

}

case class AdminCode(provName: String, cituName: String, countyName: String)


class MysqlSource extends RichSourceFunction[AdminCode] {


  var con: Connection = _

  /**
    * open 方法; 在run 方法之前执行，只执行一次
    *
    */
  override def open(parameters: Configuration): Unit = {
    //1、加载驱动
    Class.forName("com.mysql.jdbc.Driver")
    //2、建立链接
    con = DriverManager.getConnection("jdbc:mysql://master:3306/tour", "root", "123456")

  }

  /**
    * open 方法; 在run 方法之后执行，只执行一次
    */
  override def close(): Unit = {

    //关闭链接
    con.close()
  }

  /**
    * run 方法智慧运行一次
    *
    */
  override def run(ctx: SourceFunction.SourceContext[AdminCode]): Unit = {

    //查询数据
    val stat: PreparedStatement = con.prepareStatement("select prov_name,city_name,county_name from admin_code")

    //执行查询
    val resultSet: ResultSet = stat.executeQuery()

    //解析数据
    while (resultSet.next()) {
      val prov_name: String = resultSet.getString("prov_name")
      val city_name: String = resultSet.getString("city_name")
      val county_name: String = resultSet.getString("county_name")

      //发送到下游
      ctx.collect(AdminCode(prov_name, city_name, county_name))
    }

  }

  override def cancel(): Unit = {

  }
}
